package com.nl.deal;

import java.util.Iterator;
import java.util.List;

import com.alibaba.druid.support.logging.Log;
import com.alibaba.druid.support.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.druid.util.StringUtils;
import com.nl.kafka.KafkaConsumer;
import com.nl.util.AES;
import com.nl.util.StringUtil;

public class CollectData implements Runnable {
	private static final Logger log = LoggerFactory.getLogger(CollectData.class);
	private static final Logger logger1 = LoggerFactory.getLogger("kafkaData");
	private KafkaConsumer consumer;
	private String bootstrap;
	private String topic;
	int count1 = 0;
	private String groupId;
	private int count;
	public CollectData(String bootstrap,String topic,String groupId) {
		this.bootstrap = bootstrap;
		this.topic = topic;
		this.groupId = groupId;
	}
	
	private void init() {
		count =0 ;
		count1 =0 ;
		consumer = new KafkaConsumer(bootstrap,groupId);
		consumer.setBatchDurationMillis(100L);
		List<String> topics = StringUtil.split(topic, ",");
		consumer.setTopic(topics);
		consumer.connect();
	}
	@Override
	public void run() {
		init();
		log.info("start get data from kafka");
		while(true) {
			Iterator<ConsumerRecord<String, byte[]>> it = consumer.poll();
			while(it.hasNext()) {
				try{
					String msg = new String(it.next().value());
					if (!"".equals(msg) && null != msg) {
						List<String> list = StringUtil.split(msg, ",");
						String msisdn = list.get(0);
						msisdn = AES.decrypt(msisdn, "HuaDongJieSuo123");
						count1++;
						logger1.info("poll msg count:"+count1+";msg="+msg+";msisdn = "+msisdn);
						//log.error("poll msg count:"+count1+";msg="+msg+";msisdn = "+msisdn);
					}
					DataCache.dealData.add(msg);
					count++;
					if (count%100000 == 0) {
						log.info("consumer"+count);
					}
				} catch (Exception e) {
					log.error("get msg error!",e);
				}
			}
		}
	}
	
	private void close() {

		if (null != consumer) {
			consumer.close();
		}
	}

	public static void main(String[] args) throws Exception {
		String msg = "BLVMNwECs8sm2_yFBzAy5Q,3dLtwvSiAuoyjYdpN_c4wg,14,20211201111739,20211201111751,2000250,14,0,0";
		List<String> list = StringUtil.split(msg, ",");
		String msisdn = list.get(0);
		System.out.println("---"+msisdn);
		System.out.println("---"+AES.decrypt("UEZhAV0dLqgLHPqhxOHNzg", "HuaDongJieSuo123"));
	}
}
